package operators;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import pojo.PageViewCount;

public class PV_Modify_01_KeyedProcessFunction extends KeyedProcessFunction<Long, PageViewCount, PageViewCount> {

    private ValueState<Long> countValueState; // 保存pv-count值

    @Override
    public void open(Configuration parameters) {
        countValueState = getRuntimeContext().getState(
                new ValueStateDescriptor<>("page-view-count", Long.class)
        );
    }

    @Override
    public void processElement(PageViewCount value, Context ctx, Collector<PageViewCount> out) throws Exception {
        if ( countValueState.value() != null){
            countValueState.update( countValueState.value() + value.getViewCount() );
            ctx.timerService().registerEventTimeTimer(value.getTimestamp() + 1);
        }else{
            countValueState.update(value.getViewCount().longValue());
        }

    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<PageViewCount> out) throws Exception {
        out.collect(
                new PageViewCount( "pv", ctx.getCurrentKey(), countValueState.value().intValue())
        );
        countValueState.clear();
    }
}
